/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.prediction.recalc;

import aiaudio.Application;
import aiaudio.database.tables.RatingsListTable;
import aiaudio.database.tables.RatingsMatrixTable;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Vector;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;

/**
 *
 * @author nastya
 */
public class UserGroupRatingCalculationReducer extends TableReducer<ImmutableBytesWritable, UserGrouppingAggregate, ImmutableBytesWritable> {

    public static final int ARTISTS_PER_PACK = 500;
    public static final String INITIAL_RATING_MATRIX = "initialRatingMatrix";
    public static final String RECOMENDATION_SET_LIST = "recomSetList";
    public static final long TIMESTAMP_VERSION = 1L;
    HTable ratingMatrix;
    HTable recomSetList;
    String ratingMatrixName;
    Result defaultVoting;

    @Override
    protected void reduce(ImmutableBytesWritable key, Iterable<UserGrouppingAggregate> values, Context context) throws IOException, InterruptedException {
        Scan scan = new Scan();
        scan.setFilter(RatingsListTable.createByUserFilter(key.get()));
        ResultScanner scanner = recomSetList.getScanner(scan);

        Result[] res;

        while ((res = scanner.next(ARTISTS_PER_PACK)) != null && res.length > 0) {
            String[] includedArtists = new String[res.length];
            for (int i = 0; i < res.length; i++) {
                String artist = RatingsListTable.getArtist(res[i]);
                includedArtists[i] = artist;
            }
            calculateArtistGroupRating(values, includedArtists, key, context);
        }


    }

    protected void calculateArtistGroupRating(Iterable<UserGrouppingAggregate> values, String[] includedArtists, ImmutableBytesWritable key, Context context) throws InterruptedException, IOException {
        for (UserGrouppingAggregate userGrouppingAggregate : values) {
            for (String artist : includedArtists) {
                Result artistRatingList = getRatingList(artist);
                HashMap<GroupDistance, Vector<UserDistance>> map = userGrouppingAggregate.getMap();

                HashMap<Double, Double> resultedMap = new HashMap<Double, Double>();
                for (Map.Entry<GroupDistance, Vector<UserDistance>> entry : map.entrySet()) {
                    double groupDistance = entry.getKey().getDistance();
                    double calculatedGroupValue = caclulate(entry.getValue(), artistRatingList);
                    resultedMap.put(groupDistance, calculatedGroupValue);
                }
                double result = caclculatedWeightedResult(resultedMap);
                byte[] artistKey = Bytes.toBytes(artist);
                byte[] userKey = key.get();
                Put put = new Put(Bytes.add(artistKey, userKey));
                RatingsListTable.addCalculatedRating(put, result);
                context.write(new ImmutableBytesWritable(artistKey), put);
            }

            break;
        }
    }

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        super.setup(context);
        ratingMatrixName = context.getConfiguration().get(INITIAL_RATING_MATRIX);
        ratingMatrix = Application.database().getByName(ratingMatrixName).getTable();
        String recSetName = context.getConfiguration().get(RECOMENDATION_SET_LIST);
        recomSetList =  Application.database().getByName(recSetName).getTable();
        defaultVoting = setupDefaultVoting();
    }

    private Result setupDefaultVoting() throws IOException {
        return RatingsMatrixTable.getDefaultRatings(ratingMatrix);
    }

    protected Result getRatingList(String artistName) throws IOException {
        Get get = new Get(Bytes.toBytes(artistName));
        return ratingMatrix.get(get);
    }

    protected double getRatingOrDefault(String user, Result artistRatingList) {
        return RatingsMatrixTable.getArtistRatingOrDefault(artistRatingList, defaultVoting, Bytes.toBytes(user));
    }

    protected double caclulate(Vector<UserDistance> value, Result artistRatingList) {
        double sum = 0.0;
        double sumWeight = 0.0;
        for (UserDistance userDistance : value) {
            double distance = userDistance.getDistance();
            String userId = userDistance.getUserId();
            double rating = getRatingOrDefault(userId, artistRatingList);
            sum += distance * rating;
            sumWeight += distance;
        }
        return sum / sumWeight;
    }

    protected double caclculatedWeightedResult(HashMap<Double, Double> resultedMap) {
        double sum = 0;
        double sumWeight = 0;

        for (Map.Entry<Double, Double> entry : resultedMap.entrySet()) {
            Double distance = entry.getKey();
            Double groupRating = entry.getValue();

            sum += distance * groupRating;
            sumWeight += distance;
        }

        return sum / sumWeight;
    }
}
